package org.example.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * world count(dataset 不推荐)
 */
public class WordCountBatch {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> dataSource = env.readTextFile("aa");
//        FlatMapOperator<String, Tuple2<String, Integer>> w1 = dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
//            @Override
//            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
//                String[] split = s.split(" ");
//                for (String str : split) {
//                    Tuple2<String, Integer> tuple2 = Tuple2.of(str, 1);
//                    collector.collect(tuple2);
//                }
//            }
//        });

        FlatMapOperator<String, Tuple2<String, Integer>> w1 = dataSource.flatMap(
                (String s, Collector<Tuple2<String, Integer>> collector) -> {
                    String[] split = s.split(" ");
                    for (String str : split) {
                        collector.collect(Tuple2.of(str, 1));
                    }
                })
//                lamda泛型擦除问题，需要指定入参类型
                .returns(Types.TUPLE(Types.STRING,Types.INT));

        UnsortedGrouping<Tuple2<String, Integer>> groupBy = w1.groupBy(0);
        AggregateOperator<Tuple2<String, Integer>> sum = groupBy.sum(1);
        sum.print();


    }
}